package com.tc.nio.c4;

import com.tc.util.ByteBufferUtil;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Auther: tianchao
 * @Date: 2022/2/8 14:30
 * @Description:
 */
@Slf4j
public class MultiThreadServer {
    public static void main(String[] args) throws IOException {
        Thread.currentThread().setName("boss--thread");
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.configureBlocking(false);
        Selector boss = Selector.open();
        SelectionKey scKey = ssc.register(boss, 0, null);
        scKey.interestOps(SelectionKey.OP_ACCEPT);
        ssc.bind(new InetSocketAddress(8088));
        Worker[] workers = new Worker[Runtime.getRuntime().availableProcessors()];
        for (int i = 0; i < workers.length; i++) {
            workers[i] = new Worker("worker-"+i);
        }
        AtomicInteger index = new AtomicInteger();
        while (true){
            boss.select();
            final Iterator<SelectionKey> iterator = boss.selectedKeys().iterator();
            while (iterator.hasNext()){
                SelectionKey key = iterator.next();
                iterator.remove();;
                if (key.isAcceptable()){
                    final SocketChannel sc = ssc.accept();
                    log.debug("建立连接， {}", sc);
                    sc.configureBlocking(false);
                    Worker worker = workers[index.getAndIncrement() % workers.length];
                    worker.register(()->{
                        try {
                            sc.register(worker.selector, SelectionKey.OP_READ, null);
                        } catch (ClosedChannelException e) {
                            e.printStackTrace();
                        }
                    });

                }
            }
        }
    }

    static class Worker implements Runnable{
        private String name;
        private Thread thread;
        private Selector selector;
        private volatile boolean start;
        private Queue<Runnable> queue = new ConcurrentLinkedDeque<>();

        public Worker(String name) {
            this.name = name;
        }

        public void register(Runnable runnable) throws IOException {
            if (!start){
                thread = new Thread(this,name);
                selector = Selector.open();
                thread.start();
                start = true;
            }
            queue.add(runnable);
            selector.wakeup();
        }

        @Override
        public void run() {
            while (true){
                try {

                    selector.select();
                    while (!queue.isEmpty()){
                        queue.poll().run();
                    }
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    Iterator<SelectionKey> iterator = selectionKeys.iterator();
                    while (iterator.hasNext()){
                        SelectionKey selectionKey = iterator.next();
                        iterator.remove();
                        if (selectionKey.isReadable()){
                            try {
                                SocketChannel socketChannel = (SocketChannel)selectionKey.channel();
                                ByteBuffer byteBuffer = ByteBuffer.allocate(20);
                                socketChannel.read(byteBuffer);
                                byteBuffer.flip();
                                ByteBufferUtil.debugAll(byteBuffer);
                            } catch (IOException ex){
                                log.debug("read异常 client强制断开");
                                ex.printStackTrace();
                                selectionKey.cancel();
                            }
                        }
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
